package com._58city.spark.app;

import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.util.Arrays;

import org.apache.commons.lang3.StringUtils;
import org.apache.flume.source.avro.AvroFlumeEvent;
import org.apache.spark.streaming.flume.SparkFlumeEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com._58city.spark.app.dao.BaseDto;
import com._58city.spark.app.dao.CPCDisplayDto;
import com._58city.spark.app.dao.KeyInfo;

public class SparkUtil {
	
	
	public static final String ERROR = "ERROR";
	
	
	public static String event2Str(SparkFlumeEvent event){
		if(null != event){
			AvroFlumeEvent afe = event.event();
			if(null != afe){
				ByteBuffer buffer = afe.getBody();
				if(null != buffer){
					byte[] bytes = buffer.array();
					if(null != bytes){
						String content = new String(bytes);
						if(StringUtils.isNotBlank(content)){
							return content;
						}
					}
				}
			}
		}
		return null;
	}
	
	//该方法是把时间向小规整
	public static long makeFormateDate(long time,int statisticalFrequency) {
		time=time/statisticalFrequency*statisticalFrequency;
		return time;
	}
	
	public static String getMapKeyNotTime(BaseDto dto,KeyInfo keyInfo){
		String res = keyInfo.getKey_name()+"@";
		for (String key : keyInfo.getKey_values()) {
			if (key.startsWith("${") && key.endsWith("}")) {
				key = key.substring(2, key.length() - 1);
				try {
					Class<?> superClass = dto.getClass();
					Field field = null;
					while (superClass != null && field == null) {
			            try {
			                field = superClass.getDeclaredField(key);
			            } catch (NoSuchFieldException e) {
			            }
			            superClass = superClass.getSuperclass();
			        }
					field.setAccessible(true);
					key = field.get(dto).toString();
				}
				catch (Exception e) {
					return null;
				}
			}
			res += key+",";
		}
		return res.substring(0,res.length()-1);
	}
	
	public static String getMapKeyNotTime(BaseDto dto,String[] keys){
		Logger log = LoggerFactory.getLogger(SparkUtil.class);
		
		String res = "";
		for (String key : keys) {
			if (key.startsWith("${") && key.endsWith("}")) {
				key = key.substring(2, key.length() - 1);
				try {
					Class<?> superClass = dto.getClass();
					Field field = null;
					while (superClass != null && field == null) {
			            try {
			                field = superClass.getDeclaredField(key);
			            } catch (NoSuchFieldException e) {
			            }
			            superClass = superClass.getSuperclass();
			        }
					field.setAccessible(true);
					key = field.get(dto).toString();
				}
				catch (Exception e) {
					log.error("tran key error",e);
					return null;
				}
			}
			res += key+",";
		}
		return res.substring(0,res.length()-1);
	}
	
	public static void main(String[] args) {
		
		CPCDisplayDto dto = new CPCDisplayDto();
		dto.setBusiLine("err");
		dto.setPlatform("-1");
		String[] keys = new String[]{"${platform}","${busiLine}"};
		KeyInfo ki = new KeyInfo("map",keys);
		System.out.println(getMapKeyNotTime(dto, ki));
		
		String s = "map@11,22";
		String key_name = s.split("@")[0];
		String[] key_values = s.split("@")[1].split(",");
		System.out.println(key_name +" "+ Arrays.toString(key_values));
	}
	
	public static String getMapKey(BaseDto dto,String[] keys) {
		String res = "";
		String time =Long.toString(dto.getFormaterTime());
		res = time;
		for (String key : keys) {
			if (key.startsWith("${") && key.endsWith("}")) {
				key = key.substring(2, key.length() - 1);
				try {
					Class<?> superClass = dto.getClass();
					Field field = null;
					while (superClass != null && field == null) {
			            try {
			                field = superClass.getDeclaredField(key);
			            } catch (NoSuchFieldException e) {
			            }
			            superClass = superClass.getSuperclass();
			        }
					field.setAccessible(true);
					key = field.get(dto).toString();
				}
				catch (Exception e) {
					return null;
				}
			}
			res = res + "," + key;
		}
		return res;
	}

}
